package com.datacenter.service;

import com.datacenter.entity.*;
import com.datacenter.mapper.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.List;

/**
 * 数据同步服务
 * 
 * 业务含义：
 * 实现场景3的每日增量数据同步功能，每日凌晨2:00同步前一天(T-1)的增量数据
 * 包括新增资源、用户行为、资源更新等数据的同步
 */
@Slf4j
@Service
public class DataSyncService {
    
    @Autowired
    private ResourceMetadataMapper resourceMetadataMapper;
    
    @Autowired
    private ApiCallLogMapper apiCallLogMapper;
    
    @Autowired
    private DownloadLogMapper downloadLogMapper;
    
    @Autowired
    private UserApplicationMapper userApplicationMapper;
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    /**
     * 每日增量数据同步任务
     * 业务含义：每日凌晨2:00执行，同步前一天的所有增量数据
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void syncIncrementalData() {
        log.info("开始执行每日增量数据同步任务");
        
        try {
            // 计算T-1的时间范围
            LocalDateTime yesterday = LocalDateTime.now().minusDays(1);
            LocalDateTime startTime = yesterday.withHour(0).withMinute(0).withSecond(0);
            LocalDateTime endTime = yesterday.withHour(23).withMinute(59).withSecond(59);
            
            // 1. 同步新增资源元数据
            syncNewResources(startTime, endTime);
            
            // 2. 同步接口调用日志
            syncApiCallLogs(startTime, endTime);
            
            // 3. 同步下载日志
            syncDownloadLogs(startTime, endTime);
            
            // 4. 同步用户申请记录
            syncUserApplications(startTime, endTime);
            
            // 5. 同步资源更新记录
            syncResourceUpdates(startTime, endTime);
            
            log.info("每日增量数据同步任务完成");
            
        } catch (Exception e) {
            log.error("增量数据同步失败", e);
        }
    }
    
    /**
     * 同步新增资源
     * 业务含义：同步T-1新增的接口/库表/Excel元数据
     */
    private void syncNewResources(LocalDateTime startTime, LocalDateTime endTime) {
        log.info("开始同步新增资源数据，时间范围：{} 到 {}", startTime, endTime);
        
        List<ResourceMetadata> newResources = resourceMetadataMapper.selectByCreateTimeRange(startTime, endTime);
        
        for (ResourceMetadata resource : newResources) {
            // 发送到Kafka进行实时处理
            kafkaTemplate.send("resource-metadata", resource);
            log.debug("同步新增资源：{}", resource.getResourceName());
        }
        
        log.info("新增资源同步完成，共{}条记录", newResources.size());
    }
    
    /**
     * 同步接口调用日志
     * 业务含义：同步T-1的接口调用记录，用于实时监控和趋势分析
     */
    private void syncApiCallLogs(LocalDateTime startTime, LocalDateTime endTime) {
        log.info("开始同步接口调用日志，时间范围：{} 到 {}", startTime, endTime);
        
        List<ApiCallLog> callLogs = apiCallLogMapper.selectByCallTimeRange(startTime, endTime);
        
        for (ApiCallLog callLog : callLogs) {
            // 发送到Kafka进行实时处理
            kafkaTemplate.send("api-call-log", callLog);
        }
        
        log.info("接口调用日志同步完成，共{}条记录", callLogs.size());
    }
    
    /**
     * 同步下载日志
     * 业务含义：同步T-1的文件下载记录，用于使用行为分析
     */
    private void syncDownloadLogs(LocalDateTime startTime, LocalDateTime endTime) {
        log.info("开始同步下载日志，时间范围：{} 到 {}", startTime, endTime);
        
        List<DownloadLog> downloadLogs = downloadLogMapper.selectByDownloadTimeRange(startTime, endTime);
        
        for (DownloadLog downloadLog : downloadLogs) {
            // 发送到Kafka进行实时处理
            kafkaTemplate.send("download-log", downloadLog);
        }
        
        log.info("下载日志同步完成，共{}条记录", downloadLogs.size());
    }
    
    /**
     * 同步用户申请记录
     * 业务含义：同步T-1的用户申请记录，用于审批效率分析
     */
    private void syncUserApplications(LocalDateTime startTime, LocalDateTime endTime) {
        log.info("开始同步用户申请记录，时间范围：{} 到 {}", startTime, endTime);
        
        List<UserApplication> applications = userApplicationMapper.selectByApplicationTimeRange(startTime, endTime);
        
        for (UserApplication application : applications) {
            // 发送到Kafka进行实时处理
            kafkaTemplate.send("user-application", application);
        }
        
        log.info("用户申请记录同步完成，共{}条记录", applications.size());
    }
    
    /**
     * 同步资源更新记录
     * 业务含义：同步T-1修改的资源信息，如接口地址变更、库表字段调整
     */
    private void syncResourceUpdates(LocalDateTime startTime, LocalDateTime endTime) {
        log.info("开始同步资源更新记录，时间范围：{} 到 {}", startTime, endTime);
        
        List<ResourceMetadata> updatedResources = resourceMetadataMapper.selectByUpdateTimeRange(startTime, endTime);
        
        for (ResourceMetadata resource : updatedResources) {
            // 发送到Kafka进行实时处理
            kafkaTemplate.send("resource-update", resource);
            log.debug("同步资源更新：{}", resource.getResourceName());
        }
        
        log.info("资源更新记录同步完成，共{}条记录", updatedResources.size());
    }
    
    /**
     * 手动触发增量同步
     * 业务含义：支持手动触发指定时间范围的增量同步，用于数据修复
     */
    public void manualSyncIncrementalData(LocalDateTime startTime, LocalDateTime endTime) {
        log.info("手动触发增量数据同步，时间范围：{} 到 {}", startTime, endTime);
        
        syncNewResources(startTime, endTime);
        syncApiCallLogs(startTime, endTime);
        syncDownloadLogs(startTime, endTime);
        syncUserApplications(startTime, endTime);
        syncResourceUpdates(startTime, endTime);
        
        log.info("手动增量数据同步完成");
    }
}
